package com.ucex.common.rocketmq.producer.ali;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import com.aliyun.openservices.ons.api.Message;
import com.ucex.common.rocketmq.MQProperties;
import com.ucex.common.rocketmq.domain.MessageEntity;

import lombok.extern.slf4j.Slf4j;

/**
 * 消息发送处理类
 */
@SuppressWarnings("unused")
@Slf4j
public abstract class AbstractMessageProducerProcess {


    //发送消息线程数量
    private static final int DEFAULT_WORKER_SIZE = 20;

    //消息池大小相对线程数量的倍数,默认大小20 * 4
    private static final int POOL_SIZE_FACTOR = 4;
    protected LinkedBlockingQueue<Message> pool;

    public void init(MQProperties properties) {
        initPool();
        initProducer(properties);
        initExecutor();
        log.info("AliMQProducerProcess initialized.");
    }

    /**
     * 初始化内存队列，该队列是一个阻塞指定长度的队列
     */
    private void initPool() {
        int workerSize = getWorkerSize();
        int poolSize = POOL_SIZE_FACTOR * workerSize;
        log.info("Creating pool, size: {}", poolSize);
        pool = new LinkedBlockingQueue<Message>(poolSize);
    }

    private void initExecutor() {
        int workerSize = getWorkerSize();
        log.info("Creating producer workers, size: {}", workerSize);
        ExecutorService es = Executors.newFixedThreadPool(workerSize);
        for (int i = 0; i < workerSize; i++) {
            es.submit(new Runnable() {
                public void run() {
                    while (!Thread.currentThread().isInterrupted()) {
                        sendMessageFromQueue();
                    }
                }
            });
        }
        es.shutdown();
    }

    
    private int getWorkerSize() {
        String size = null;
        if (null == size || size.trim().isEmpty()) {
            return DEFAULT_WORKER_SIZE;
        }

        try {
            return Integer.valueOf(size);
        } catch (NumberFormatException e) {
            return DEFAULT_WORKER_SIZE;
        }
    }

    private boolean isEmpty(String topic) {
        return null == topic || 0 == topic.trim().length() || "null".equals(topic);
    }

    /**
     * 初始化消息发送者信息
     */
    public abstract void initProducer(MQProperties properties);

    /**
     * 从消息队列获取数据并且发送,当消息队列没数据的时候会阻塞
     */
    public abstract void sendMessageFromQueue();

    /**
     * 消息推送到消息队列中，可能会根据不同的MQ组装成不同的消息格式
     * @param <T>
     * @param t
     * @param <T> 发送的消息对象
     */
    public abstract void pushMessageToQueue(MessageEntity t);
    
}
